package cn.doitedu.rtdw.data_etl;

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

@Slf4j
public class Etl03_AccessTimeAnalyse {
    public static void main(String[] args) throws Exception {

        // 创建编程入口
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3);
        env.setParallelism(1);
        // 状态后端
        env.setStateBackend(new HashMapStateBackend());
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 创建连接器，读取kafka中的dwd行为明细
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setTopics("dwd-events-detail")
                .setGroupId("gp002")
                .setStartingOffsets(OffsetsInitializer.latest())
                //.setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.LATEST))
                .setBootstrapServers("doitedu:9092")
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        DataStreamSource<String> ds = env.fromSource(source, WatermarkStrategy.noWatermarks(), "dwd-kfk");

        SingleOutputStreamOperator<DwdUserEvent> beanStream = ds.map(json -> JSON.parseObject(json, DwdUserEvent.class));

        // 按同一个人的同一次会话，进行keyby
        KeyedStream<DwdUserEvent, Tuple2<Long, String>> keyedStream = beanStream.keyBy(new KeySelector<DwdUserEvent, Tuple2<Long, String>>() {
            @Override
            public Tuple2<Long, String> getKey(DwdUserEvent eventBean) throws Exception {
                return Tuple2.of(eventBean.getUser_id(), eventBean.getSession_id());
            }
        });

        // 核心逻辑： 虚拟事件插值  （api开发）
        SingleOutputStreamOperator<DwdUserEvent> virtualEventStream = keyedStream.process(new KeyedProcessFunction<Tuple2<Long, String>, DwdUserEvent, DwdUserEvent>() {
            ValueState<DwdUserEvent> beanState;
            ValueState<Long> timerState;

            @Override
            public void open(Configuration parameters) throws Exception {
                beanState = getRuntimeContext().getState(new ValueStateDescriptor<DwdUserEvent>("bean", DwdUserEvent.class));
                timerState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("timerState", Long.class));
            }

            @Override
            public void processElement(DwdUserEvent eventBean, KeyedProcessFunction<Tuple2<Long, String>, DwdUserEvent, DwdUserEvent>.Context ctx, Collector<DwdUserEvent> out) throws Exception {

                // 当接收到一个用户行为时，先删除此前注册的状态清理定时器
                Long preTimerTime = timerState.value();
                if (preTimerTime != null) ctx.timerService().deleteProcessingTimeTimer(preTimerTime);
                // 然后注册一个新的定时器
                ctx.timerService().registerProcessingTimeTimer(ctx.timerService().currentProcessingTime() + 5000);

                DwdUserEvent stateEventBean = beanState.value();
                if (stateEventBean == null) {
                    // 对本次收到的数据，填充 on_page及page_start_time 字段，并放入状态
                    eventBean.setOn_page(eventBean.getProperties().get("url"));
                    eventBean.setPage_start_time(eventBean.getEvent_time());
                    beanState.update(eventBean);

                    stateEventBean = beanState.value();
                }

                // 判断本次的数据，是否需要更新所在页面及页面的起始打开时间
                if (eventBean.getEvent_id().equals("page_load")) {

                    // 输出一条虚拟事件（事件的时间要更新成此刻最新时间）
                    stateEventBean.setEvent_time(eventBean.getEvent_time());
                    out.collect(stateEventBean); // 输出虚拟事件

                    // 更新之前状态中记录的 所在页面，及页面的起始打开时间
                    stateEventBean.setOn_page(eventBean.getProperties().get("url"));
                    stateEventBean.setPage_start_time(eventBean.getEvent_time());

                    out.collect(stateEventBean); // 输出本次实际事件

                } else if (eventBean.getEvent_id().equals("wake_up")) {

                    // 更新一下页面起始时间为此刻的事件时间即可
                    stateEventBean.setEvent_time(eventBean.getEvent_time());
                    stateEventBean.setPage_start_time(eventBean.getEvent_time());
                    out.collect(stateEventBean); // 输出本次实际事件

                } else {
                    // 其他事件，只需要将状态数据中的事件时间更新成本次收到的事件时间即可
                    stateEventBean.setEvent_time(eventBean.getEvent_time());
                    out.collect(stateEventBean);

                    // 如果是app_close事件，则应该在输出结果数据后清除掉状态数据
                    if (eventBean.getEvent_id().equals("app_close")) {
                        beanState.clear();
                    }

                }
            }


            @Override
            public void onTimer(long timestamp, KeyedProcessFunction<Tuple2<Long, String>, DwdUserEvent, DwdUserEvent>.OnTimerContext ctx, Collector<DwdUserEvent> out) throws Exception {
                // 清除存放bean的状态
                log.error("爱死你了,状态终于要被清理了,此刻时间为: {} ",timestamp);
                beanState.clear();
            }
        });


        //virtualEventStream.print();

        // 核心逻辑：时间窗口聚合
        tenv.createTemporaryView("tmp",virtualEventStream,
                Schema.newBuilder()
                        .column("user_id", DataTypes.BIGINT())
                        .column("device_type", DataTypes.STRING())
                        .column("release_channel", DataTypes.STRING())
                        .column("session_id", DataTypes.STRING())
                        .column("page_type", DataTypes.STRING())
                        .column("on_page", DataTypes.STRING())
                        .column("page_start_time", DataTypes.BIGINT())

                        .column("event_time", DataTypes.BIGINT())
                        .columnByExpression("rt","to_timestamp_ltz(event_time,3)")
                        .watermark("rt","rt - interval '0' second")
                .build());


        // 创建连接器，对接doris，并写出
        tenv.executeSql(
                " CREATE TABLE res_doris (         "
                        +" dt  DATE,                     "
                        +" user_id BIGINT,               "
                        +" device_type STRING,           "
                        +" release_channel STRING,       "
                        +" session_id STRING,            "
                        +" page_type STRING,             "
                        +" page_url STRING,              "
                        +" page_start_time BIGINT,       "
                        +" page_end_time BIGINT          "
                        + " ) WITH (                               "
                        + "    'connector' = 'doris',              "
                        + "    'fenodes' = 'doitedu:8030',         "
                        + "    'table.identifier' = 'dws.acc_timelong_page',  "
                        + "    'username' = 'root',                "
                        + "    'password' = '',                    "
                        + "    'sink.label-prefix' = 'doris_label_dws"+System.currentTimeMillis()+"'"
                        + " )                                         ");


        tenv.executeSql(
                " INSERT INTO res_doris SELECT                                    "
                        +"   TO_DATE(DATE_FORMAT(page_start_time,'yyyy-MM-dd')) as dt,  "
                        +"   user_id,                                                "
                        +"   device_type,                                            "
                        +"   release_channel,                                        "
                        +"   session_id,                                             "
                        +"   page_type,                                              "
                        +"   on_page as page_url,                                    "
                        +"   page_start_time,                                        "
                        +"   max(event_time) as page_end_time                        "
                        +"                                                           "
                        +" FROM TABLE(                                               "
                        +"   TUMBLE(TABLE tmp,DESCRIPTOR(rt),INTERVAL '5' MINUTE)    "
                        +" )                                                         "
                        +" GROUP BY                                                  "
                        +"   window_start,                                           "
                        +"   window_end,                                             "
                        +"   user_id,                                                "
                        +"   device_type,                                            "
                        +"   release_channel,                                        "
                        +"   session_id,                                             "
                        +"   page_type,                                              "
                        +"   on_page,                                                "
                        +"   page_start_time                                         "
        );

        env.execute();
    }
}
